package club.vann.mock.kafka;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

/** Kafka消费中心，负责整体项目的Kafka消息消费。 */
@Component
@Log4j2
public class KafkaConsumer {
  /** K: topic; V: Comsumer#ConsumerRecord */
  private static final Map<String, Consumer<ConsumerRecord<String, byte[]>>>
      KAFKA_CONSUMER_MAPPING = new ConcurrentHashMap<>();

  /** 注册 Consumer 回调方法 */
  public static void registerConsumer(
      String topic, Consumer<ConsumerRecord<String, byte[]>> consumerRecord) {
    KAFKA_CONSUMER_MAPPING.put(topic, consumerRecord);
  }

  /** 消费 Biyi Acceptor Request 消息 */
  @org.springframework.kafka.annotation.KafkaListener(topics = "biyi.acceptor.input")
  public void consumerIndex(ConsumerRecord<String, byte[]> consumerRecord) {
    consumer(consumerRecord);
  }

  /** 回调方法 */
  private void consumer(ConsumerRecord<String, byte[]> consumerRecord) {
    String topic = consumerRecord.topic();
    Consumer<ConsumerRecord<String, byte[]>> consumerRecordConsumer =
        KAFKA_CONSUMER_MAPPING.get(topic);
    if (consumerRecordConsumer == null) {
      return;
    }
    consumerRecordConsumer.accept(consumerRecord);
  }
}
